package org.lzfto.flink.demo.business.flinkTest.flinkObjects.source.clickdemo;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;

import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.core.io.InputStatus;
import org.lzfto.flink.demo.business.flinkTest.entity.ClickEvent;
import org.lzfto.flink.demo.business.flinkTest.entity.ClickEventSourceConfig;
import org.lzfto.flink.demo.utils.TimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;

/**
 * 用随机算法生成一定量的数据，并且部分的数据可能会超出窗口
 * @author lzfto
 * @创建时间 2025年7月9日 11:46:12
 */
public class ClickEventSourceReader implements SourceReader<ClickEvent, ClickEventSplit> {

	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	private final List<ClickEventSplit> assignedSplits = Collections.synchronizedList(new ArrayList<>());
	private CompletableFuture<Void> availableFuture = new CompletableFuture<>();
	private final BlockingQueue<ClickEvent> clickEventQueue = new LinkedBlockingQueue<>(1000);
	private volatile boolean isWorking;	
	private volatile int dataQty=0;  //生成的数据
	private volatile int pollQty=0;  //取出的数据
    
	
	//属性
	private final ClickEventSourceConfig config;

	public ClickEventSourceReader(ClickEventSourceConfig config) {
		this.config=config;
		this.isWorking = true;
		LOGGER.info("创建点击事件生成器,测试持续时间{}秒", this.config.testElapseSecs());
	}

	@Override
	public void close() throws Exception {
		LOGGER.info("退出点击事件生成器");
	}

	@Override
	public void start() {
		Date startTime = new Date();
		long lastTime = startTime.getTime() + config.testElapseSecs() * 1000;
		LOGGER.info("开始制造数据!开始时间:{},应该停止时间:{}", TimeUtil.formatDate(startTime),
				TimeUtil.formatDate(new Date(lastTime)));
		ThreadLocalRandom random = ThreadLocalRandom.current();
		//构造可能延时数据的逻辑
		//窗口长度+允许延迟=窗口能够容忍的最迟时间
		//窗口长度+允许延迟+额外延时时间=可能超出窗口容忍的最迟时间
		int maxLateness=config.getMaxLateness();
		// 创建一个线程并执行数据生成任务
		new Thread(
				() -> {
					LOGGER.info("开始子线程....");
					try {
						dataQty = 0;
						while (true) {
							long currentTimeValue = (new Date()).getTime();
							String item = String.valueOf(random.nextInt(1, 6));
							
							//随机偏移介于 [0,maxLateness] 之间
							//如此可能生成超出窗口的数据
							int ranOffset = random.nextInt(0, maxLateness+1);
							
							// event的timestamp随机设置为比当前时间少1~15秒，以便模拟超时窗口测试效果。							
							ClickEvent event = new ClickEvent(item, item, currentTimeValue - 1000 * ranOffset);
							clickEventQueue.offer(event);
							dataQty++;
							if (currentTimeValue >= lastTime) {
								break;
							}
							Thread.sleep(250);
						}
						LOGGER.info("本分片生成数据:{}条", dataQty);
						
                        //对clickEventQueue的数据根据item分组统计数量
						//使用迭代器访问BlockingQueue中的元素
						Iterator<ClickEvent> iterator = clickEventQueue.iterator();
						Map<String, Long> itemCountMap = new HashMap<>();
						while (iterator.hasNext()) {
							ClickEvent event = iterator.next();
							Long count = itemCountMap.get(event.getItemId());
							if (count == null) {
								itemCountMap.put(event.getItemId(), 1L);
							} else {
								itemCountMap.put(event.getItemId(), count + 1);
							}
						}
						LOGGER.info("统计结果:{}", JSON.toJSONString(itemCountMap,Feature.PrettyFormat));
						isWorking = false;
					} catch (InterruptedException e) {
						LOGGER.error("生成随机的点击事件失败:{}", e.getMessage());
					}
				}).start();

	}

	@Override
	public InputStatus pollNext(ReaderOutput<ClickEvent> output) throws Exception {
		ClickEvent event = this.clickEventQueue.poll();
		if (event != null) {
			output.collect(event);
			// 提示还有数据可以取
			synchronized (this) {
				availableFuture = new CompletableFuture<>();
			}
			this.pollQty++;
			return InputStatus.MORE_AVAILABLE;
		}
		
		if(isWorking) {
			//表示现在还在工作
			return InputStatus.MORE_AVAILABLE; 
		}
		else {
			//数据已经就绪
			if (this.pollQty<this.dataQty) {
				synchronized (this) {
					availableFuture = new CompletableFuture<>();
				}
				return InputStatus.MORE_AVAILABLE; 
			}
			else {
				return InputStatus.END_OF_INPUT;
			}
			
		}
	}

	@Override
	public List<ClickEventSplit> snapshotState(long checkpointId) {
		return new ArrayList<>(assignedSplits);
	}

	@Override
	public CompletableFuture<Void> isAvailable() {
		/**
		 * 特别说明：
		 * 1.这个方法通常最开始的时候不会被调用。
		 * 2.当 pollNext 被调用，并返回非MORE_AVAILABLE的时候，Flink会尝试调用本方法
		 * 3.如果本方法获得可用信息，那么Flink会继续调用pollNext
		 * 4.如果Flink通过本方法的返回，发现不可用，那么就不会再调用pollNext了。
		 * 
		 * 所以，一般的做法是：初始的时候设置new CompletableFuture<>()，并不断在pollNext中创建
		 */
		
		LOGGER.info("check available---");
		return availableFuture;
	}

	@Override
	public void addSplits(List<ClickEventSplit> splits) {
		if (splits == null || splits.isEmpty()) {
			LOGGER.warn("未分配新的 splits");
		} else {
			assignedSplits.addAll(splits);
			// 接收到新的分片
			int splitQty = 0;
			for (ClickEventSplit split : splits) {
				LOGGER.info("接收到新的点击事件分片:{}", split.splitId());
				splitQty++;
			}
			LOGGER.info("累计获得分片数量:{}", splitQty);
		}
	}

	@Override
	public void notifyNoMoreSplits() {
		LOGGER.info("接收到 noMoreSplits 通知,不会有新的分片来了！");
	}

}
